package com.youxin.dataSet.api;

import com.youxin.dataStream.pojo.WordCount;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.operators.DataSource;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;

/**
 * 广播变量的使用
 * 注意：广播变量目的是一个机器上只缓存一份数据
 */
public class DataSetBroadCast {
    public static void main(String[] args) throws Exception {
        //获取运行环境
        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();

        //1：准备需要广播的数据
        ArrayList<Tuple2<String, Integer>> broadData = new ArrayList<>();
        broadData.add(new Tuple2<>("zs",18));
        broadData.add(new Tuple2<>("ls",20));
        broadData.add(new Tuple2<>("ww",17));
        DataSet<Tuple2<String, Integer>> tupleData = env.fromCollection(broadData).setParallelism(1);

        //1.1:处理需要广播的数据,把数据集转换成map类型，map中的key就是用户姓名，value就是用户年龄
        DataSet<HashMap<String, Integer>> toBroadcast = tupleData.map(new MapFunction<Tuple2<String, Integer>, HashMap<String, Integer>>() {
            @Override
            public HashMap<String, Integer> map(Tuple2<String, Integer> value) throws Exception {
                HashMap<String, Integer> res = new HashMap<>();
                res.put(value.f0, value.f1);
                return res;
            }
        });

        //源数据
        DataSource<String> data = env.fromElements("zs", "ls", "ww","zs", "ls", "ww").setParallelism(3);

        //注意：在这里需要使用到RichMapFunction获取广播变量
        DataSet<String> ret = data.map(new RichMapFunction<String, String>() {
            //为了缓存数据创建的集合   //怀疑这样写法有问题，这样还是每个线程都又有了一份数据，是不是需要把数据直接广播过来，不需要再做其他的操作？？？
            List<HashMap<String,Integer>> broadCastMap = new ArrayList<>();
            //这个变量可能有问题
            HashMap<String,Integer> allMap = new HashMap<>();

            WordCount wordCount = new WordCount();
            @Override
            public void open(Configuration parameters) throws Exception {
                super.open(parameters);
                //获取广播变量的数据
                this.broadCastMap = getRuntimeContext().getBroadcastVariable("broadCastMapName");
                System.out.println("wc地址:"+wordCount);
                for(HashMap map:broadCastMap){
                    allMap.putAll(map);
                    System.out.println(Thread.currentThread().getName()+":"+map);
                    System.out.println("allMap的数据:"+allMap);
                    Thread.sleep(1000);
                }
            }

            @Override
            public String map(String s) throws Exception {
                Integer age = allMap.get(s);
                return s+":"+age;
            }
        }).withBroadcastSet(toBroadcast,"broadCastMapName");//把数据广播到算子中

        ret.print();

    }
}
